package worker

import (
	"context"
	"crontab/common"
	"go.etcd.io/etcd/api/v3/mvccpb"
	clientv3 "go.etcd.io/etcd/client/v3"
	"log"
	"time"
)

// 单例

type EtcdJob struct {
	EtcdClient *clientv3.Client
	EtcdKv clientv3.KV
	EtcdLease clientv3.Lease
	EtcdWatcher clientv3.Watcher
}

var  (
	G_EtcdJob *EtcdJob
)

// 监听etcd的任务
func (etcd *EtcdJob) WatchJob() {
	var (
		getRes *clientv3.GetResponse
		err error
		job *common.Job
		jobEvent *common.JobEvent
	)
	// 获取到所有的key,初始化目前的所有任务推送到调度协程
	getRes,err = etcd.EtcdKv.Get(context.TODO(),common.ETCD_KEY_PREFIX,clientv3.WithPrefix())
	for _, kv := range getRes.Kvs {
		if job,err = common.UnPackJob(kv.Value); err == nil {
			// TODO job同步给调度协程
			jobEvent = common.BuildJobEvent(common.JOB_EVENT_TYPE_SAVE,job)
			//fmt.Println(jobEvent)
			G_Scheduler.PushJobEvent(jobEvent)
		}
		continue
	}

	// 由目前的状态开始往后监听 取版本号 +1
	// Revision
	//: 作用域为集群，逻辑时间戳，全局单调递增，任何 key 修改都会使其自增
	//CreateRevision
	//: 作用域为 key, 等于创建这个 key 时的 Revision, 直到删除前都保持不变
	//ModRevision
	//: 作用域为 key, 等于修改这个 key 时的 Revision, 只要这个 key 更新都会改变
	//Version
	//: 作用域为 key, 某一个 key 的修改次数(从创建到删除)，与以上三个 Revision 无关
	go func() {
		var (
			startRevision int64
			watchChan clientv3.WatchChan
			watchResp clientv3.WatchResponse
			event *clientv3.Event
			jobName string
			jobEvent *common.JobEvent
		)
		// 从此版本开始监听
		startRevision = getRes.Header.Revision + 1
		watchChan = etcd.EtcdWatcher.Watch(context.TODO(),common.ETCD_KEY_PREFIX,clientv3.WithRev(startRevision),clientv3.WithPrefix())
		for watchResp = range watchChan {
			for _, event = range watchResp.Events {
				switch event.Type {
				case mvccpb.PUT:
					// 保存事件 -> 推送到调度协程
					if job,err = common.UnPackJob(event.Kv.Value);err != nil {
						log.Printf("解析etcd的job失败 " + err.Error())
						continue
					}

					// 构建一个新增Event
					jobEvent = common.BuildJobEvent(common.JOB_EVENT_TYPE_SAVE,job)

				case mvccpb.DELETE:
					// 删除事件 -> 删除事件需要jobName
					// 删除需要知道删除的任务名称
					jobName = common.GetJobNameByKey(event.Kv.Key)
					job = &common.Job{JobName: jobName}
					// 构建一个删除Event
					jobEvent = common.BuildJobEvent(common.JOB_EVENT_TYPE_DEL,job)
				}

				// jobEvent 推送至调度协程
				//fmt.Println(jobEvent)
				G_Scheduler.PushJobEvent(jobEvent)
			}
		}

	}()

}

// 监听强杀目录
func (etcd *EtcdJob) WatchKills() {
	var (
		job *common.Job
	)
	go func() {
		var (
			watchChan clientv3.WatchChan
			watchResp clientv3.WatchResponse
			event *clientv3.Event
			jobName string
			jobEvent *common.JobEvent
		)
		watchChan = etcd.EtcdWatcher.Watch(context.TODO(),common.ETCD_KILL_KEY_PREFIX,clientv3.WithPrefix())
		for watchResp = range watchChan {
			for _, event = range watchResp.Events {
				switch event.Type {
				case mvccpb.PUT:
					// 强杀
					jobName = common.GetKillJobNameByKey(event.Kv.Key)
					job = &common.Job{JobName:jobName }
					// 构建一个新增Event
					jobEvent = common.BuildJobEvent(common.JOB_EVENT_TYPE_KILL,job)
					// jobEvent 推送至调度协程
					//fmt.Println(jobEvent)
					G_Scheduler.PushJobEvent(jobEvent)
				case mvccpb.DELETE:
					// 删除是租约到期的删除,无需关系
				}


			}
		}

	}()

}

func InitEtcd() (err error) {
	var (
		config *clientv3.Config
		etcdClient *clientv3.Client
		kv clientv3.KV
		lease clientv3.Lease
		watcher clientv3.Watcher
	)
	config = &clientv3.Config{
		Endpoints: G_Config.ApiEndpoints,
		DialTimeout: time.Second * time.Duration(G_Config.ApiEtcdTimeout),
		Username: G_Config.EtcdUserName,
		Password: G_Config.EtcdPassword,
	}
	etcdClient,err = clientv3.New(*config)
	if err != nil {
		return err
	}

	kv = clientv3.NewKV(etcdClient)

	lease = clientv3.NewLease(etcdClient)

	watcher = clientv3.NewWatcher(etcdClient)


	G_EtcdJob = &EtcdJob{
		EtcdClient: etcdClient,
		EtcdKv: kv,
		EtcdLease: lease,
		EtcdWatcher:watcher,
	}

	return nil
}





